﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using MinistryStandardPlatformBusiness.Helper;
using MinistryStandardPlatformBusiness.MessagePool.QueueSession;
using Newtonsoft.Json;
using ZeroMQ;

namespace MinistryStandardPlatformBusiness.MessagePool
{
    public class MessageQueueHander
    {
        private MessagePublish msgPub;
        ZmqContext BusinessZmqContext;
        ZmqSocket BusinessZmqSocket;
        ZmqContext ClientZmqContext;
        ZmqSocket ClientZmqSocket;
        ZmqContext ClientSubZmqContext;
        ZmqSocket ClientSubZmqSocket;
        System.Threading.Thread MsgQueueBusinessThread;//业务中心线程
        System.Threading.Thread ClientSubThread;//双向通信线程
        System.Timers.Timer onLineUserHanderTimer;//在线用户的定时器维护
        System.Threading.Thread businessHeartThread;//业务中心心跳线程
        int websequnce = 1;
        int OnlineUserMaxCache;
        Frame GidFrame;

        public MessageQueueHander()
        {
           // startListen();
        }
        public MessageQueueHander(string strBusinessEndPoint, string strClientLocalPoint, string strClientPubSubPoint,
            int MaxQueLen = 1024000, string LogPath = "D:\\Log\\", string WebNumSequnce = "1", string OnlineUserMax = "10")
        {
            UserSession.BusinessEndPoint = strBusinessEndPoint;
            UserSession.ClientLocalPoint = strClientLocalPoint;
            UserSession.ClientPubSubPoint = strClientPubSubPoint;
            UserSession.MaxQueLen = MaxQueLen > 1024000 ? MaxQueLen : 1024000;
            UserSession.LogPath = LogPath;
            websequnce = WebNumSequnce.Length > 0 ? int.Parse(WebNumSequnce) : 1;
            OnlineUserMaxCache = OnlineUserMax.Length > 0 ? int.Parse(OnlineUserMax) : 10;

            initIdFrame();//初始化全局变量
            LogPathConfig();//Log日志路径配置
            initCatch();//实时缓存
            startListen();//启动监听
        }
        public void startListen()
        {
            onLineUserHanderTimer = new System.Timers.Timer(OnlineUserMaxCache * 60 * 1000);//10分钟
            onLineUserHanderTimer.Elapsed += new System.Timers.ElapsedEventHandler(theOut);
            onLineUserHanderTimer.AutoReset = true;//一直执行
            onLineUserHanderTimer.Enabled = true;//执行事件

            msgPub = new MessagePublish();
            msgPub.Publish += msgPub_Publish;//订阅消息推送至客户端
            msgPub.businessPublish += msgBusiness_Publish;//订阅消息分发事件至业务中心

            initClientSubHander();

            initMsgQueueBusinessHander();
            initClientMsgQueueHander();

            MsgQueueBusinessThread = new System.Threading.Thread(MsgQueueBusinessListener);
            MsgQueueBusinessThread.IsBackground = true;
            MsgQueueBusinessThread.Name = "MsgQueueBusinessThread";
            MsgQueueBusinessThread.Start();

            //initTestHander();
            //ClientMsgQueueThread = new System.Threading.Thread(MsgQueueSubListener);
            //ClientMsgQueueThread.IsBackground = true;
            //ClientMsgQueueThread.Start();

            ClientSubThread = new System.Threading.Thread(MsgQueueSubGetListener);
            ClientSubThread.IsBackground = true;
            ClientSubThread.Name = "ClientSubThread";
            ClientSubThread.Start();

            
        }
        #region initSocket
        /// <summary>
        /// 业务中心服务器连接
        /// </summary>
        private void initMsgQueueBusinessHander()
        {
            BusinessZmqContext = ZmqContext.Create();
            BusinessZmqSocket = BusinessZmqContext.CreateSocket(SocketType.DEALER);
            BusinessZmqSocket.SendHighWatermark = UserSession.MaxQueLen;
            BusinessZmqSocket.ReceiveHighWatermark = UserSession.MaxQueLen;
        }
       /// <summary>
       /// web消息订阅及分发
       /// </summary>
        private void initClientMsgQueueHander()
        {

            ClientZmqContext = ZmqContext.Create();
            ClientZmqSocket = ClientZmqContext.CreateSocket(SocketType.PUB);
            //ClientZmqSocket.Bind(UserSession.ClientLocalPoint);
            ClientZmqSocket.SendHighWatermark = UserSession.MaxQueLen;
        }
        /// <summary>
        /// web cs 等双向通信
        /// </summary>
        private void initClientSubHander()
        {

            if (ClientSubZmqContext == null)
            {
                ClientSubZmqContext = ZmqContext.Create();
                ClientSubZmqSocket = ClientSubZmqContext.CreateSocket(SocketType.REP);
              //  ClientSubZmqSocket.Bind(UserSession.ClientPubSubPoint);
            }
        }
        #endregion
        
        #region  消息监听
        /// <summary>
        /// 监听来至业务中心的消息
        /// </summary>
        private void MsgQueueBusinessListener()
        {
            BusinessZmqSocket.Connect(UserSession.BusinessEndPoint);
            #region
            byte[] id = new byte[2];
            id[0] = (byte)websequnce;
            id[1] = 0;
            Frame idFrame = new Frame(id);
            TimeSpan timeOut = new TimeSpan(0,0,0,0,200);
            ZmqMessage msg = new ZmqMessage();
            msg.Append(idFrame);
            msg.AppendEmptyFrame();
            #endregion
            BusinessZmqSocket.SendMessage(msg);
            //initBusinessHeartThread();
            while (true)
            {
                Log.WriteLog("gpsLocationStart", "start");
                try
                {
                    ZmqMessage message = BusinessZmqSocket.ReceiveMessage(timeOut);
                    if (message.Count() > 0)
                    {
                        foreach (var f in message)
                        {
                            msgPub.RecieveMsg(Encoding.UTF8.GetString(f));
                        }
                    }
                    #region
                    else
                    {
                        //businesszmqsocket.sendmessage(heartmsg);
                        Log.WriteLog("gpslocationnull", ":0");
                    }
                    #endregion
                }
                catch (Exception e)
                {
                    Log.WriteLog("gpsLocationCError", e.ToString());
                }
                #region
                //ZmqMessage msg = new ZmqMessage();
                //msg.Append(idFrame);
                //msg.Append(data);
                BusinessZmqSocket.SendMessage(msg);
                #endregion
                System.Threading.Thread.Sleep(50);
            }
        }
        /// <summary>
        /// 监听来至业务中心的实时消息1 测试
        /// </summary>
        private void MsgQueueListener()
        {
            using (var ctx = ZmqContext.Create())
            {
                using (ZmqSocket rep = ctx.CreateSocket(SocketType.DEALER))
                {
                    rep.Connect(UserSession.BusinessEndPoint);

                    byte[] id = new byte[2];
                    //id[0] = (byte)'1';
                    id[0] = (byte)websequnce;
                    id[1] = 0;
                    byte[] data = new byte[2];
                    data[0] = (byte)'w';
                    data[1] = 0;
                    Frame idFrame = new Frame(id);
                    int index = 0;
                    ZmqMessage heartmsg = new ZmqMessage();
                    heartmsg.AppendEmptyFrame();
                    rep.SendMessage(heartmsg);
                    while (true)
                    {
                        ZmqMessage message = rep.ReceiveMessage();
                        if (message.Count() > 0)
                        {
                            foreach (var f in message)
                            {
                                Log.WriteLog("BufferSize", f.BufferSize.ToString());
                                msgPub.RecieveMsg(Encoding.UTF8.GetString(f));
                                //Log.WriteLog("gpsLocation",index.ToString()+ ":" + Encoding.UTF8.GetString(f));
                            }
                        }
                        ZmqMessage msg = new ZmqMessage();
                        msg.Append(idFrame);
                        msg.Append(data);
                        rep.SendMessage(msg);
                        index++;
                        System.Threading.Thread.Sleep(200);
                    }
                }
            }
        }

        /// <summary>
        /// 客户端与WEB的双向通信
        /// </summary>
        private void MsgQueueSubGetListener()
        {
            Log.WriteLog("MsgFromClientStart", "start");
            string temp = "";
            while (true)
            {
                string sendTag = ClientSubZmqSocket.Receive(Encoding.UTF8);//接收
               temp = msgPub.HeartMsgFromClientHander(sendTag);
                Log.WriteLog("MsgFromClient", sendTag);
                ClientSubZmqSocket.Send(temp, Encoding.UTF8);//发送
                System.Threading.Thread.Sleep(50);
            }
        }
        #endregion

        #region  web至客户端的消息分发及数据处理
        /// <summary>
        /// 实时消息分发
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        void msgPub_Publish(object sender, PubEventArgs e)
        {
            MsgQueuePubHander(e);
        }
        private void MsgQueuePubHander(PubEventArgs e)
        {
            if (ClientZmqSocket == null)
            {
                initClientMsgQueueHander();//初始化异常处理
            }
            //2014/06/23 从else中移出
            ClientZmqSocket.SendMore(e.sim, Encoding.UTF8);
            ClientZmqSocket.Send(e.body, Encoding.UTF8);
            
            Log.WriteLog("sendDetails", e.body);

        }

        #endregion
        #region web至业务中心数据分发

        void msgBusiness_Publish(object sender,PubEventArgs e)
        {
            if (BusinessZmqContext == null)
            {
                initMsgQueueBusinessHander();
                BusinessZmqSocket.Connect(UserSession.BusinessEndPoint);
            }

            ZmqMessage msg = new ZmqMessage();
            msg.Append(GidFrame);
            msg.Append(Encoding.UTF8.GetBytes(e.body));
            BusinessZmqSocket.SendMessage(msg);

            Log.WriteLog("Business"+e.type, e.body);
        }
       
        #endregion
        #region 在线用户维护
        /// <summary>
        /// 在线用维护
        /// </summary>
        /// <param name="source"></param>
        /// <param name="e"></param>
        private void theOut(object source,System.Timers.ElapsedEventArgs e)
        {
            UserSession.onLineUserHander();

        }
        #endregion

        #region 初始化配置
        private void LogPathConfig()
        {
            try
            {
                if (!System.IO.Directory.Exists(UserSession.LogPath))
                {
                    System.IO.Directory.CreateDirectory(UserSession.LogPath);
                }
            }
            catch (Exception e)
            {
                System.IO.Directory.CreateDirectory("D:\\LogH\\");
                Log.WriteLog("Error",e.ToString());
            }
        }
        private void initIdFrame()
        {
            byte[] id = new byte[2];
            id[0] = (byte)websequnce;
            id[1] = 0;
            //byte[] data = new byte[2];
            //data[0] = (byte)'w';
            //data[1] = 0;
            GidFrame = new Frame(id);         
        }

        private void initBusinessHeartThread()
        {
            ZmqMessage msg = new ZmqMessage();
            msg.Append(GidFrame);

            businessHeartThread = new System.Threading.Thread(() =>
            {

                if (BusinessZmqContext == null)
                {
                    initMsgQueueBusinessHander();
                    BusinessZmqSocket.Connect(UserSession.BusinessEndPoint);
                }
                BusinessZmqSocket.SendMessage(msg);
                Log.WriteLog("businessHeartThread", GidFrame.ToString());
                System.Threading.Thread.Sleep(500);

            });
            businessHeartThread.Name = "businessHeartThread";
            businessHeartThread.IsBackground = true;
            businessHeartThread.Start();
        }


        private void initCatch()
        {
            System.Threading.Tasks.Task catchTask = new Task(() =>
            {
               // UserSession.initGpsLocationCacheOld();//从GPS数据库初始化缓存
                UserSession.initGpsLocationCache();//从基础数据库缓存信息初始化缓存
                UserSession.isCouludCatchWorked = true;
            });
            catchTask.Start();
        }
        #endregion

    }
}